package ex.ex2;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * 舆情内容长度小于12个字，极端舆情
 */
public class LengthBoltC extends BaseBasicBolt {
    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        String sentence = input.getStringByField("sentence");
        if (sentence.replace(" 电话舆情", "").length() <= 12) {
            collector.emit(new Values(input.getValue(0), input.getStringByField("sentence") + " 极端舆情。"));
        } else {
            collector.emit(new Values(input.getValue(0), input.getStringByField("sentence") + "。"));
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("id", "info"));
    }
}
